// readers/tdengine_reader.go
package readers

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"mig2kwdb/common"
	"strings"

	_ "github.com/taosdata/driver-go/v3/taosRestful"
	_ "github.com/taosdata/driver-go/v3/taosWS"
)

type TDengineReader struct {
	db *sql.DB
}

func NewTDengineReader(connInfo common.DBConnectionInfo) (*TDengineReader, error) {
	// 格式: dsn = "root:taosdata@ws(10.20.50.89:6041)/"

	var protocol string = "ws"
	// 构建基础连接字符串
	dsn := fmt.Sprintf("%s:%s@%s(%s:%d)/",
		connInfo.User,
		connInfo.Password,
		protocol,
		connInfo.Ip,
		connInfo.Port)

	// 如果指定了数据库名，则添加到连接字符串中
	if connInfo.DatabaseName != "" {
		dsn += "/" + connInfo.DatabaseName
	}

	// 添加查询参数以优化连接
	dsn += "?charset=utf8&timeout=30s"

	//log.Printf("TDengine DSN: %s", dsn)
	//dsn = "root:taosdata@ws(10.20.50.89:6041)/"
	db, err := sql.Open("taosWS", dsn)
	if err != nil {
		return nil, fmt.Errorf("failed to open TDengine connection: %v", err)
	}

	// 设置连接池参数
	db.SetMaxIdleConns(5)
	db.SetMaxOpenConns(10)

	// 测试连接
	if err := db.Ping(); err != nil {
		db.Close()
		return nil, fmt.Errorf("failed to ping TDengine: %v", err)
	}

	return &TDengineReader{
		db: db,
	}, nil
}

// GetDB 返回 TDengine 数据库连接
func (r *TDengineReader) GetDB() *sql.DB {
	return r.db
}

// Close 关闭 TDengine 数据库连接
func (r *TDengineReader) Close() {
	if r.db != nil {
		r.db.Close()
	}
}
func (r *TDengineReader) GetDataBaseType() string {
	return "tdengine"
}
func (r *TDengineReader) IsNetworkError(err error) bool {
	return false
}

// GetDataBases retrieves a list of all databases in the TDengine instance
func (r *TDengineReader) GetDataBases() ([]string, error) {
	query := "SHOW DATABASES"

	rows, err := r.db.Query(query)
	if err != nil {
		log.Println(query, err)
		return nil, err
	}
	defer rows.Close()

	var databases []string
	for rows.Next() {
		var dbName string

		// 第一个字段是数据库名
		err := rows.Scan(&dbName)
		if err != nil {
			return nil, err
		}

		// 排除系统数据库
		if dbName != "log" && dbName != "information_schema" && dbName != "performance_schema" && dbName != "audit" {
			databases = append(databases, dbName)
		}
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return databases, nil
}

// GetDatabaseTables retrieves a list of all tables in the specified database
func (r *TDengineReader) GetDatabaseTables(databaseName, schemaName string) ([]string, error) {

	query := "SHOW " + databaseName + ".STABLES"
	stableRows, err := r.db.Query(query)
	if err != nil {
		return nil, err
	}
	defer stableRows.Close()

	var tables []string

	// 收集超级表名
	for stableRows.Next() {
		var stableName string

		err := stableRows.Scan(&stableName)
		if err != nil {
			return nil, err
		}
		tables = append(tables, stableName)
	}

	return tables, nil
}
func (r *TDengineReader) MapDataTypes(column []common.KwdbColumn) []common.KwdbColumn {
	for i := range column {
		switch strings.ToUpper(column[i].Type) {
		case "REAL":
			column[i].Type = common.DataTypes.FLOAT8.Name

		default:
		}
	}
	return column

}

// GetDatabaseKeepProperty retrieves the keep property (data retention period) for a TDengine database
func (r *TDengineReader) getDatabaseKeepProperty(databaseName string) (string, error) {
	// 查询数据库的keep属性
	query := fmt.Sprintf("SELECT `name`,`keep` FROM information_schema.ins_databases WHERE name = '%s'", databaseName)

	rows, err := r.db.Query(query)
	if err != nil {
		return "", fmt.Errorf("failed to query database properties: %v", err)
	}
	defer rows.Close()

	var dbName, keepValue string
	if rows.Next() {
		err = rows.Scan(&dbName, &keepValue)
		if err != nil {
			return "", fmt.Errorf("failed to scan database properties: %v", err)
		}
		//30d,30d,30d
		values := strings.Split(keepValue, ",")
		if len(values) > 0 {
			keepValue = values[0]
		}
		return keepValue, nil
	}

	return "", fmt.Errorf("database %s not found", databaseName)
}

func (r *TDengineReader) GetTableDefinition(databaseName, schemaName, tableName string) (common.TableDefinition, error) {
	var err error
	var def common.TableDefinition
	def.DatabaseName = databaseName
	def.SchemaName = schemaName
	def.TableName = tableName
	def.Columns, err = r.GetTableColumn(databaseName, schemaName, tableName)
	if err != nil {
		return def, err
	}
	def.Properties = make(map[string]interface{})
	def.Properties["keep"], _ = r.getDatabaseKeepProperty(databaseName)
	return def, nil
}

// GetTableColumn retrieves column information for a specific table in a TDengine database
func (r *TDengineReader) GetTableColumn(databaseName, schemaName, tableName string) ([]common.KwdbColumn, error) {

	query := "DESCRIBE " + databaseName + "." + tableName
	rows, err := r.db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var columns []common.KwdbColumn
	cols, err := rows.Columns()
	if err != nil {
		return nil, err
	}
	value_source := make([]interface{}, len(cols))
	for i := range value_source {
		value_source[i] = &sql.NullString{}
	}
	for rows.Next() {

		// TDengine DESCRIBE 返回: field type length note encode compress level
		err := rows.Scan(value_source...)
		if err != nil {
			return nil, err
		}
		var column common.KwdbColumn
		for i, col := range cols {
			switch strings.ToLower(col) {
			case "field":
				column.Name = value_source[i].(*sql.NullString).String
			case "note":
				column.Comment = *value_source[i].(*sql.NullString)
				if strings.ToUpper(column.Comment.String) == "TAG" {
					column.PrimaryKey.String = "YES"
					column.PrimaryKey.Valid = true
					column.Tag = "YES"
				}
			case "type":
				column.Type = value_source[i].(*sql.NullString).String
				if strings.ToUpper(column.Type) == "TIMESTAMP" {
					column.PrimaryKey.String = "YES"
					column.PrimaryKey.Valid = true
				}
			case "length":
				column.Length = *value_source[i].(*sql.NullString)

			}
		}

		columns = append(columns, column)
	}

	if err = rows.Err(); err != nil {
		return nil, err
	}

	return r.MapDataTypes(columns), nil
}
func (r *TDengineReader) GetConditionRows(databaseName, schemaName, tableName string, columns []common.KwdbColumn, condition string) (common.RowsInterface, error) {
	orderby := ""
	keyNum := 0
	sql := "select "
	for idx, col := range columns {
		if idx > 0 {
			sql += ","
		}
		sql += " " + (col.Name)
		if col.PrimaryKey.String == "YES" {
			if keyNum > 0 {
				orderby += ","
			}
			orderby += " " + (col.Name)
			keyNum++
		}
	}
	sql += " from " + (databaseName) + "." + (tableName)
	if condition != "" {
		sql += " where " + condition
	}
	if orderby != "" {
		sql += " order by " + orderby
	}

	rows, err := r.db.Query(sql)
	return common.WrapSQLRows(rows), err
}

func (r *TDengineReader) GetStartPos() (common.ReplicateStartPos, error) {
	return common.ReplicateStartPos{}, nil
}
func (r *TDengineReader) SetDataBases(dataBases []common.DatabaseConfig) {

}
func (r *TDengineReader) SetReplicateHandler(handler common.ReplicateHandlerInterface) {
}
func (r *TDengineReader) StartReplicate(ctx context.Context, startPos common.ReplicateStartPos) error {
	return nil
}

// TDEngineRows 自定义Rows结构，用于分批获取TDengine数据
type TDEngineRows struct {
	reader       *TDengineReader
	databaseName string
	tableName    string
	batchSize    int
	currentRows  *sql.Rows
	offset       int
	finished     bool
	columns      []string
	condition    string // 添加条件字段
	orderBy      string // 添加排序字段
}

// Next 获取下一批数据
func (tr *TDEngineRows) Next() bool {
	// 如果当前批次已经处理完
	if tr.currentRows == nil || !tr.currentRows.Next() {
		// 如果已经完成所有数据的获取，返回false
		if tr.finished {
			return false
		}

		// 关闭当前批次
		if tr.currentRows != nil {
			tr.currentRows.Close()
		}

		// 构建下一批数据的查询语句
		sql := fmt.Sprintf("SELECT * FROM %s.%s", tr.databaseName, tr.tableName)
		if tr.condition != "" {
			sql += " WHERE " + tr.condition
		}
		if tr.orderBy != "" {
			sql += " ORDER BY " + tr.orderBy
		}
		sql += fmt.Sprintf(" LIMIT %d OFFSET %d", tr.batchSize, tr.offset)

		rows, err := tr.reader.db.Query(sql)
		if err != nil {
			log.Printf("Error querying next batch: %v", err)
			tr.finished = true
			return false
		}

		tr.currentRows = rows
		tr.offset += tr.batchSize

		// 检查是否还有更多数据
		if !rows.Next() {
			tr.finished = true
			return false
		}

		// 重新定位到第一行数据
		// 注意：这里我们需要重新组织逻辑，因为上面的Next()已经消费了第一行
		return true
	}

	return true
}

// Scan 扫描当前行数据
func (tr *TDEngineRows) Scan(dest ...interface{}) error {
	if tr.currentRows == nil {
		return fmt.Errorf("no rows available")
	}
	return tr.currentRows.Scan(dest...)
}

// Columns 返回列名
func (tr *TDEngineRows) Columns() ([]string, error) {
	if tr.currentRows == nil {
		return nil, fmt.Errorf("no rows available")
	}
	if tr.columns == nil {
		cols, err := tr.currentRows.Columns()
		if err != nil {
			return nil, err
		}
		tr.columns = cols
	}
	return tr.columns, nil
}

// Close 关闭资源
func (tr *TDEngineRows) Close() error {
	if tr.currentRows != nil {
		return tr.currentRows.Close()
	}
	return nil
}

// Err 返回错误信息
func (tr *TDEngineRows) Err() error {
	if tr.currentRows == nil {
		return nil
	}
	return tr.currentRows.Err()
}
